//package com.tiancheng.trade.order.util.cmq.queue.consumer;
//
//import com.tiancheng.trade.order.util.cmq.model.QueueMessage;
//import com.tiancheng.trade.order.util.cmq.queue.PullConsumerFactory;
//import com.tiancheng.trade.order.util.cmq.queue.QueueConsumer;
//import com.fasterxml.jackson.core.type.TypeReference;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import org.apache.commons.collections4.CollectionUtils;
//import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.BeansException;
//import org.springframework.beans.factory.BeanFactory;
//import org.springframework.beans.factory.BeanFactoryAware;
//import org.springframework.beans.factory.DisposableBean;
//import org.springframework.beans.factory.InitializingBean;
//
//import java.util.*;
//
///**
// * @Author: likailun
// * @Description:
// * @Date: create in 2021/11/19 5:33 下午
// */
//public abstract class AbstractCustomQueueConsumer implements BeanFactoryAware, InitializingBean, DisposableBean, QueueConsumer {
//	/**
//	 * 参考下：https://blog.csdn.net/tiantiantbtb/article/details/140834967
//	 * https://rocketmq.apache.org/zh/docs/4.x/consumer/03pull
//	 * https://blog.csdn.net/meser88/article/details/130780676
//	 * https://blog.csdn.net/tiantiantbtb/article/details/140834967
//	 */
//    private BeanFactory beanFactory;
//	PullConsumerFactory consumerFactory;
//	private ObjectMapper objectMapper;
//    /**
//     * 连接池是否关闭
//     */
//    volatile boolean isCloseConnectionPool = false;
//
//    @Override
//    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
//        this.beanFactory = beanFactory;
//    }
//
//    @Override
//    public void afterPropertiesSet() throws Exception {
//        this.consumerFactory = this.beanFactory.getBean(PullConsumerFactory.class);
//        String queueName = this.getQueueName();
//        if (queueName == null || queueName.isEmpty()) {
//            throw new IllegalArgumentException("队列名称不能为空");
//        }
//        Integer pollingWaitSeconds = this.getPollingWaitSeconds();
//        if (pollingWaitSeconds != null) {
//            if (pollingWaitSeconds < 0 || pollingWaitSeconds > 30) {
//                throw new IllegalArgumentException("pollingWaitSeconds取值范围为0到30");
//            }
//        }
//        if (this.getHandleThreadNum() <= 0) {
//            throw new IllegalArgumentException("消息处理线程数必须大于0");
//        }
//
//        MessageFetch messageFetch = new MessageFetch();
//        for (int i = 0; i < this.getHandleThreadNum(); i++) {
//            Thread thread = new Thread(messageFetch);
//            thread.setName("queueConsumer-" + this.getQueueName() + i);
//            thread.start();
//        }
//    }
//
//    @Override
//    public void destroy() throws Exception {
//        this.isCloseConnectionPool = true;
//    }
//
//    public abstract void handle(QueueMessage message) throws Exception;
//
//    /**
//     * 拉取消息的间隔时间
//     *
//     * @throws Exception
//     */
//    public int getPullInterval() {
//        return new Random().nextInt(100);
//    }
//
//	public int getHandleThreadNum(){
//		return 1;
//	}
//	public int getPullBatchSize(){
//		return 1;
//	}
//
//    private class MessageFetch implements Runnable {
//        private final Logger logger = LoggerFactory.getLogger(MessageFetch.class);
//        private TreeMap<String, Object> fetchParam;
//
//        private DefaultLitePullConsumer pullConsumer;
//        private String queueName;
//
//        MessageFetch() {
//            this.queueName = AbstractCustomQueueConsumer.this.getQueueName();
//            this.fetchParam = new TreeMap<>();
//            this.fetchParam.put("queueName", this.queueName);
//            if (AbstractCustomQueueConsumer.this.getPollingWaitSeconds() != null) {
//                this.fetchParam.put("pollingWaitSeconds", AbstractCustomQueueConsumer.this.getPollingWaitSeconds());
//            }
//            this.pullConsumer = AbstractCustomQueueConsumer.this.consumerFactory.getPullConsumer(this.queueName);
//			this.pullConsumer.setPullBatchSize(AbstractCustomQueueConsumer.this.getPullBatchSize());
//			this.pullConsumer.setAutoCommit(false);
//        }
//
//        @Override
//        public void run() {
//            Random roRandom = new Random();
//			try {
//				this.pullConsumer.start();
//			} catch (MQClientException e) {
//				logger.error("消费者启动失败！");
//				throw new RuntimeException(e);
//			}
//			// 连接没被关闭的情况下进行循环
//            while (!Thread.currentThread().isInterrupted() && !AbstractCustomQueueConsumer.this.isCloseConnectionPool) {
//                try {
//                    Thread.sleep(AbstractCustomQueueConsumer.this.getPullInterval());
//                } catch (InterruptedException e) {
//                    Thread.currentThread().interrupt();
//                    e.printStackTrace();
//                }
//                if (!AbstractCustomQueueConsumer.this.instanceIsUp()) {
//                    continue;
//                }
//                List<QueueMessage> queueMessages = this.fetch();
//                if (CollectionUtils.isNotEmpty(queueMessages)) {
//                    try {
//						for (QueueMessage queueMessage : queueMessages) {
//							AbstractCustomQueueConsumer.this.handle(queueMessage);
//						}
//                        this.pullConsumer.commit();
//                    } catch (Exception e) {
//                        this.logger.error("cmq queue message handle failure", e);
//                    }
//                }
//            }
//        }
//
//
//        /**
//         * 删除消息
//         */
//        private void deleteMessage(QueueMessage queueMessage) {
//            //this.cmqRequest.DeleteMessage(this.queueName, queueMessage.getReceiptHandle());
//        }
//
//        private List<QueueMessage> fetch() {
//            try {
//				final List<MessageExt> messageExts = this.pullConsumer.poll();
//				if (CollectionUtils.isEmpty(messageExts)) {
//                    return Collections.emptyList();
//                }
//				final List<QueueMessage> queueMessages = new ArrayList<>();
//				for (MessageExt messageExt : messageExts) {
//					final QueueMessage queueMessage = objectMapper.readValue(messageExt.getBody(), new TypeReference<QueueMessage>() {
//					});
//					queueMessages.add(queueMessage);
//
//				}
//                return queueMessages;
//            } catch (Exception e) {
//                this.logger.error("cmq queue message fetch failure", e);
//                return null;
//            }
//        }
//    }
//
//    /**
//     * 判断实例是否是UP状态
//     */
//    boolean instanceIsUp() {
//        boolean result = true;
//        /*if (this.beanFactory.containsBean("eurekaClient")) {
//            EurekaClient eurekaClient = this.beanFactory.getBean(EurekaClient.class);
//            //如果实例的本地状态或者在eureka的状态不是up,则认为不可执行
//            try {
//                if (eurekaClient.getApplicationInfoManager().getInfo().getStatus() != InstanceInfo.InstanceStatus.UP
//                        || eurekaClient.getInstanceRemoteStatus() != InstanceInfo.InstanceStatus.UP) {
//                    result = false;
//                }
//            } catch (Exception e) {
//                result = false;
//            }
//        }*/
//        return result;
//    }
//}
